package com.jzl.canal_consumer.bean;


import com.alibaba.otter.canal.protocol.CanalEntry;
import com.jzl.util.CollectionUtil;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.alibaba.otter.canal.protocol.CanalEntry.EventType.*;

/**
 * 对同一张表
 * 以DDL语句为分隔线
 * 对DDL后面到下一个DDL语句前面之间的DML语句进行合并
 *
 * @author ZhenWuWang
 */
public class MergeSqlEntry
{
    private final Map<String, List<DmlEntry>> idDMLEntriesMap;

    private DdlEntry ddlEntry;

    private MergeSqlEntry()
    {
        idDMLEntriesMap = new HashMap<>();
    }

    public static MergeSqlEntry create()
    {
        return new MergeSqlEntry();
    }

    public DdlEntry getDdlEntry()
    {
        return ddlEntry;
    }

    public void setDdlEntry(DdlEntry ddlEntry)
    {
        this.ddlEntry = ddlEntry;
    }

    public void addDmlEntries(Map<String, List<DmlEntry>> map)
    {
        if (map == null || map.isEmpty())
        {
            return;
        }

        map.forEach((pk, dmlEntries) ->
        {
            if (dmlEntries != null && dmlEntries.size() > 0)
            {
                List<DmlEntry> list = idDMLEntriesMap.computeIfAbsent(pk, k -> new ArrayList<>());
                list.addAll(dmlEntries);
            }
        });
    }

    public Map<CanalEntry.EventType, List<DmlEntry>> merge()
    {
        //对同一个sql语句进行的values进行合并
        Map<CanalEntry.EventType, List<DmlEntry>> map = new HashMap<>();
        for (Map.Entry<String, List<DmlEntry>> entry : idDMLEntriesMap.entrySet())
        {
            List<DmlEntry> dmlEntries = CollectionUtil.filterNull(entry.getValue());
            if (dmlEntries == null || dmlEntries.isEmpty())
            {
                continue;
            }
            DmlEntry dmlEntry = merge(dmlEntries);
            if (dmlEntry == null)
            {
                continue;
            }
            List<DmlEntry> list = map.computeIfAbsent(dmlEntry.getEventType(), k -> new ArrayList<>());
            list.add(dmlEntry);
        }
        return map;
    }


    /**
     * 合并算法对同一个Id记录日志进行合并
     * . insert + insert -> insert (数据迁移+数据增量场景)
     * 2. insert + update -> insert  (update字段合并到insert)
     * 3. insert + delete -> delete
     * 4. update + insert -> insert (数据迁移+数据增量场景)
     * 5. update + update -> update
     * 6. update + delete -> delete
     * 7. delete + insert -> insert
     * 8. delete + update -> update (数据迁移+数据增量场景)
     * 9. delete + delete -> delete
     */
    private DmlEntry merge(List<DmlEntry> dmlEntries)
    {
        dmlEntries = CollectionUtil.filterNull(dmlEntries);
        if (dmlEntries == null || dmlEntries.isEmpty())
        {
            return null;
        }
        return dmlEntries.stream().reduce((dmlEntry1, dmlEntry2) ->
        {
            if (dmlEntry1.getEventType() == INSERT || dmlEntry1.getEventType() == UPDATE)
            {
                switch (dmlEntry2.getEventType())
                {
                    case INSERT:
                    case DELETE:
                        return dmlEntry2;
                    case UPDATE:
                        return mergeColumnValues(dmlEntry1, dmlEntry2);
                    default:
                        break;
                }
            } else if (dmlEntry1.getEventType() == DELETE)
            {
                return dmlEntry2;
            }
            return dmlEntry1;
        }).orElse(null);
    }

    /**
     * 合并列变量,把第二个参数的ColumnValues覆盖到第一个参数ColumnValues
     */
    private DmlEntry mergeColumnValues(DmlEntry dmlEntry1, DmlEntry dmlEntry2)
    {
        List<CanalEntry.Column> columns = new ArrayList<>(dmlEntry1.getRowData().getAfterColumnsList());
        CanalEntry.RowData.Builder builder = dmlEntry1.getRowData().toBuilder();
        Map<String, CanalEntry.Column> map = dmlEntry2
                .getRowData()
                .getAfterColumnsList()
                .stream()
                .collect(Collectors.toMap(CanalEntry.Column::getName, column -> column));
        for (int i = 0; i < columns.size(); i++)
        {
            CanalEntry.Column column = columns.get(i);
            CanalEntry.Column c = map.get(column.getName());
            if (c != null)
            {
                CanalEntry.Column newColumn = column.toBuilder().setValue(c.getValue()).build();
                builder.setAfterColumns(i, newColumn);
            }
        }
        dmlEntry1.setRowData(builder.build());
        return dmlEntry1;
    }
}
